Method: createNativeRepository(DriverConfiguration, String)

1: /**
2: * Copyright (C) 2022 Czech Technical University in Prague
3: *
4: * This program is free software: you can redistribute it and/or modify it under
5: * the terms of the GNU General Public License as published by the Free Software
6: * Foundation, either version 3 of the License, or (at your option) any
7: * later version.
8: *
9: * This program is distributed in the hope that it will be useful, but WITHOUT
10: * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
11: * FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
12: * details. You should have received a copy of the GNU General Public License
13: * along with this program. If not, see <http://www.gnu.org/licenses/>.
14: */
15: package cz.cvut.kbss.ontodriver.sesame.connector;
16:
17: import cz.cvut.kbss.ontodriver.config.DriverConfiguration;
18: import cz.cvut.kbss.ontodriver.sesame.config.SesameConfigParam;
19: import cz.cvut.kbss.ontodriver.sesame.exceptions.RepositoryCreationException;
20: import cz.cvut.kbss.ontodriver.sesame.exceptions.RepositoryNotFoundException;
21: import cz.cvut.kbss.ontodriver.sesame.exceptions.SesameDriverException;
22: import org.eclipse.rdf4j.model.Model;
23: import org.eclipse.rdf4j.model.Resource;
24: import org.eclipse.rdf4j.model.vocabulary.RDF;
25: import org.eclipse.rdf4j.repository.Repository;
26: import org.eclipse.rdf4j.repository.RepositoryException;
27: import org.eclipse.rdf4j.repository.config.RepositoryConfig;
28: import org.eclipse.rdf4j.repository.config.RepositoryConfigException;
29: import org.eclipse.rdf4j.repository.config.RepositoryConfigSchema;
30: import org.eclipse.rdf4j.repository.manager.RemoteRepositoryManager;
31: import org.eclipse.rdf4j.repository.manager.RepositoryManager;
32: import org.eclipse.rdf4j.repository.manager.RepositoryProvider;
33: import org.eclipse.rdf4j.repository.sail.SailRepository;
34: import org.eclipse.rdf4j.repository.sail.config.SailRepositoryConfig;
35: import org.eclipse.rdf4j.rio.RDFFormat;
36: import org.eclipse.rdf4j.rio.Rio;
37: import org.eclipse.rdf4j.sail.config.SailImplConfig;
38: import org.eclipse.rdf4j.sail.inferencer.fc.SchemaCachingRDFSInferencer;
39: import org.eclipse.rdf4j.sail.inferencer.fc.config.SchemaCachingRDFSInferencerConfig;
40: import org.eclipse.rdf4j.sail.memory.MemoryStore;
41: import org.eclipse.rdf4j.sail.nativerdf.config.NativeStoreConfig;
42: import org.slf4j.Logger;
43: import org.slf4j.LoggerFactory;
44:
45: import java.io.*;
46: import java.net.URI;
47: import java.util.Optional;
48: import java.util.Set;
49:
50: class RepositoryConnectorInitializer {
51:
52: private static final Logger LOG = LoggerFactory.getLogger(RepositoryConnectorInitializer.class);
53:
54: private static final String[] KNOWN_REMOTE_SCHEMES = {"http", "https", "ftp"};
55: private static final String LOCAL_NATIVE_REPO = "repositories/";
56: private static final String FILE_SCHEME = "file";
57: private static final String CLASSPATH_PREFIX = "classpath:";
58:
59: private final DriverConfiguration configuration;
60: private final int maxReconnectAttempts;
61:
62: private RepositoryManager manager;
63: private Repository repository;
64:
65: RepositoryConnectorInitializer(DriverConfiguration configuration, int maxReconnectAttempts) {
66: this.configuration = configuration;
67: this.maxReconnectAttempts = maxReconnectAttempts;
68: }
69:
70: void initializeRepository() throws SesameDriverException {
71: final URI serverUri = configuration.getStorageProperties().getPhysicalURI();
72: LOG.debug("Initializing connector to repository at {}", serverUri);
73: try {
74: final boolean isRemote = isRemoteRepository(serverUri);
75: if (isRemote) {
76: this.repository = connectToRemoteRepository(serverUri.toString());
77: } else {
78: this.repository = createLocalRepository();
79: }
80: verifyRepositoryCreated(serverUri, isRemote);
81: repository.init();
82: } catch (RepositoryException | RepositoryConfigException e) {
83: throw new SesameDriverException("Failed to acquire sesame repository connection.", e);
84: }
85: }
86:
87: private static boolean isRemoteRepository(URI uri) {
88: final String scheme = uri.getScheme();
89: for (String s : KNOWN_REMOTE_SCHEMES) {
90: if (s.equals(scheme)) {
91: return true;
92: }
93: }
94: return false;
95: }
96:
97: private Repository connectToRemoteRepository(String repoUri) {
98: this.manager = RepositoryProvider.getRepositoryManagerOfRepository(repoUri);
99: final RemoteRepositoryManager remoteManager = (RemoteRepositoryManager) manager;
100: final String username = configuration.getStorageProperties().getUsername();
101: if (username != null) {
102: final String password = configuration.getStorageProperties().getPassword();
103: remoteManager.setUsernameAndPassword(username, password);
104: }
105: return connectToRemote(repoUri, 1);
106: }
107:
108: private Repository connectToRemote(String repoUri, int attempts) {
109: try {
110: return manager.getRepository(RepositoryProvider.getRepositoryIdOfRepository(repoUri));
111: } catch (RepositoryException e) {
112: if (attempts < maxReconnectAttempts) {
113: LOG.warn("Unable to connect to repository {}. Error is: {}. Retrying...", repoUri, e.getMessage());
114: return connectToRemote(repoUri, attempts + 1);
115: }
116: LOG.error("Threshold of failed connection attempts reached, throwing exception.");
117: throw e;
118: }
119: }
120:
121: private Repository createLocalRepository() {
122: if (configuration.isSet(SesameConfigParam.REPOSITORY_CONFIG)) {
123: return createRepositoryFromConfig();
124: }
125: final URI localUri = configuration.getStorageProperties().getPhysicalURI();
126: if (!isFileUri(localUri) && configuration.is(SesameConfigParam.USE_VOLATILE_STORAGE)) {
127: return createInMemoryRepository();
128: } else {
129: return createNativeRepository(configuration, localUri.toString());
130: }
131: }
132:
133: private Repository createRepositoryFromConfig() {
134: LOG.trace("Creating local repository from repository config file.");
135: final RepositoryConfig repoConfig = loadRepositoryConfig();
136: this.manager = RepositoryProvider.getRepositoryManager(getRepositoryManagerBaseDir().orElse(""));
137: manager.addRepositoryConfig(repoConfig);
138: return manager.getRepository(getRepositoryId());
139: }
140:
141: private RepositoryConfig loadRepositoryConfig() {
142: try (final InputStream is = getConfigFileContent()) {
143: final Model configModel = Rio.parse(is, "", RDFFormat.TURTLE);
144: final Set<Resource> resources =
145: configModel.filter(null, RDF.TYPE, RepositoryConfigSchema.REPOSITORY).subjects();
146: assert resources.size() == 1;
147: return RepositoryConfig.create(configModel, resources.iterator().next());
148: } catch (IOException e) {
149: throw new RepositoryCreationException("Unable to create repository from the specified configuration.", e);
150: }
151: }
152:
153: private InputStream getConfigFileContent() {
154: final String configPath = configuration.getProperty(SesameConfigParam.REPOSITORY_CONFIG);
155: LOG.trace("Loading repository configuration file content from {}.", configPath);
156: if (configPath.startsWith(CLASSPATH_PREFIX)) {
157: final InputStream is =
158: getClass().getClassLoader().getResourceAsStream(configPath.substring(CLASSPATH_PREFIX.length()));
159: if (is == null) {
160: throw new RepositoryCreationException(
161: "Unable to find repository configuration file on classpath location " + configPath);
162: }
163: return is;
164: } else {
165: try {
166: return new FileInputStream(configPath);
167: } catch (FileNotFoundException e) {
168: throw new RepositoryCreationException("Unable to find repository configuration file at " + configPath,
169: e);
170: }
171: }
172: }
173:
174: private Optional<String> getRepositoryManagerBaseDir() {
175: final String physicalUri = configuration.getStorageProperties().getPhysicalURI().toString();
176: final String[] tmp = physicalUri.split(LOCAL_NATIVE_REPO);
177: return tmp.length == 2 ? Optional.of(tmp[0]) : Optional.empty();
178: }
179:
180: private String getRepositoryId() {
181: final String physicalUri = configuration.getStorageProperties().getPhysicalURI().toString();
182: final String[] tmp = physicalUri.split(LOCAL_NATIVE_REPO);
183: if (tmp.length != 2) {
184: return physicalUri;
185: }
186: String repoId = tmp[1];
187: // Get rid of the trailing slash if necessary
188: return repoId.charAt(repoId.length() - 1) == '/' ? repoId.substring(0, repoId.length() - 1) : repoId;
189: }
190:
191: private static boolean isFileUri(URI uri) {
192: return uri.getScheme() != null && uri.getScheme().equals(FILE_SCHEME);
193: }
194:
195: /**
196: * Creates a local in-memory Sesame repository which is disposed when the VM shuts down.
197: */
198: private Repository createInMemoryRepository() {
199: LOG.trace("Creating local in-memory repository.");
200: final MemoryStore ms = new MemoryStore();
201: if (configuration.is(SesameConfigParam.USE_INFERENCE)) {
202: return new SailRepository(new SchemaCachingRDFSInferencer(ms));
203: } else {
204: return new SailRepository(ms);
205: }
206: }
207:
208: /**
209: * Creates native repository.
210: * <p>
211: * This kind of repository stores data in files and is persistent after the VM shuts down.
212: */
213: private Repository createNativeRepository(DriverConfiguration configuration, String localUri) {
214: LOG.trace("Creating local native repository at " + localUri);
215: validateNativeStorePath(localUri);
216: try {
217: this.manager = RepositoryProvider.getRepositoryManagerOfRepository(localUri);
218: final String repoId = getRepositoryId();
219: final RepositoryConfig cfg = createLocalNativeRepositoryConfig(repoId, configuration);
220: manager.addRepositoryConfig(cfg);
221: return manager.getRepository(repoId);
222: } catch (RepositoryConfigException | RepositoryException e) {
223: throw new RepositoryCreationException("Unable to create local repository at " + localUri, e);
224: }
225: }
226:
227: private static void validateNativeStorePath(String path) {
228: if (path.split(LOCAL_NATIVE_REPO).length != 2) {
229: throw new RepositoryCreationException(
230: "Unsupported local RDF4J/Sesame repository path. Expected file://path/repositories/id but got " +
231: path);
232: }
233: }
234:
235: private static RepositoryConfig createLocalNativeRepositoryConfig(String repoId,
236: DriverConfiguration configuration) {
237: SailImplConfig backend = new NativeStoreConfig();
238: if (configuration.is(SesameConfigParam.USE_INFERENCE)) {
239: backend = new SchemaCachingRDFSInferencerConfig(backend);
240: }
241: final SailRepositoryConfig repoType = new SailRepositoryConfig(backend);
242: return new RepositoryConfig(repoId, repoType);
243: }
244:
245: private void verifyRepositoryCreated(URI serverUri, boolean isRemote) {
246: if (repository == null) {
247: if (isRemote) {
248: throw new RepositoryNotFoundException("Unable to reach repository at " + serverUri);
249: } else {
250: throw new RepositoryCreationException("Unable to create local repository at " + serverUri);
251: }
252: }
253: }
254:
255: RepositoryManager getManager() {
256: return manager;
257: }
258:
259: Repository getRepository() {
260: return repository;
261: }
262: }